from kafka import KafkaProducer
from kafka.admin import KafkaAdminClient
import json
import time

# 创建Kafka生产者
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],  # Kafka服务器地址
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),  # 消息序列化
    key_serializer=lambda k: k.encode('utf-8') if k else None, # 键序列化

    # SASL认证配置
    security_protocol='SASL_PLAINTEXT',  # 使用SASL_PLAINTEXT协议
    sasl_mechanism='PLAIN',              # 使用PLAIN认证机制
    sasl_plain_username='admin',  # 替换为你的用户名
    sasl_plain_password='admin123'   # 替换为你的密码
)


def validate_kafka_auth():
    """使用AdminClient快速验证认证信息"""
    try:
        admin_client = KafkaAdminClient(
            bootstrap_servers=['localhost:9092'],
            security_protocol='SASL_PLAINTEXT',
            sasl_mechanism='PLAIN',
            sasl_plain_username='admin',
            sasl_plain_password='admin123',
            request_timeout_ms=3000
        )

        # 尝试列出topics，这会触发认证
        topics = admin_client.list_consumer_groups()
        admin_client.close()
        print("Kafka认证成功！")
        return True

    except Exception as e:
        print(f"Kafka认证失败: {e}")
        return False


# 发送消息
def send_message():
    topic = 'monitor-data-topic'

    # 发送简单消息
    message = {
        'id': 1,
        'name': 'Zhang San',
        'timestamp': time.time()
    }

    # 同步发送
    future = producer.send(topic, value=message, key='user_1')
    result = future.get(timeout=10)
    print(f"消息发送成功: {result}")

    # 异步发送
    producer.send(topic, value=message)

    # 确保所有消息都已发送
    producer.flush()


if __name__ == "__main__":
    if validate_kafka_auth():
        send_message()
    producer.close()
